package yunjiao.javatutorials.guava.concurrent.comprehensive;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 高性能消息队列
 *
 * @author yangyunjiao
 */
public class MonitoredMessageQueue<T> {
    private final Lock lock = new ReentrantLock();
    private final ComprehensiveCondition notEmpty;
    private final ComprehensiveCondition notFull;
    private final Queue<T> queue;
    private final int capacity;
    private volatile boolean shutdown = false;

    public MonitoredMessageQueue(int capacity) {
        this.capacity = capacity;
        this.queue = new LinkedList<>();
        this.notEmpty = new ComprehensiveCondition(lock.newCondition());
        this.notFull = new ComprehensiveCondition(lock.newCondition());
    }

    public boolean offer(T message, long timeout, TimeUnit unit) throws InterruptedException {
        if (shutdown) {
            throw new IllegalStateException("队列已关闭");
        }

        lock.lock();
        try {
            long nanosTimeout = unit.toNanos(timeout);

            while (queue.size() == capacity && !shutdown) {
                if (nanosTimeout <= 0) {
                    System.out.println("offer 操作超时");
                    return false;
                }
                nanosTimeout = notFull.awaitNanos(nanosTimeout);
            }

            if (shutdown) {
                return false;
            }

            boolean wasEmpty = queue.isEmpty();
            queue.offer(message);

            if (wasEmpty) {
                notEmpty.signal();
            }

            System.out.printf("消息已添加，队列大小: %d/%d%n", queue.size(), capacity);
            return true;

        } finally {
            lock.unlock();
        }
    }

    public T poll(long timeout, TimeUnit unit) throws InterruptedException {
        lock.lock();
        try {
            long nanosTimeout = unit.toNanos(timeout);

            while (queue.isEmpty() && !shutdown) {
                if (nanosTimeout <= 0) {
                    System.out.println("poll 操作超时");
                    return null;
                }
                nanosTimeout = notEmpty.awaitNanos(nanosTimeout);
            }

            if (shutdown && queue.isEmpty()) {
                return null;
            }

            T message = queue.poll();
            boolean wasFull = queue.size() == capacity - 1;

            if (wasFull) {
                notFull.signal();
            }

            System.out.printf("消息已取出，队列大小: %d/%d%n", queue.size(), capacity);
            return message;

        } finally {
            lock.unlock();
        }
    }

    public void shutdown() {
        lock.lock();
        try {
            shutdown = true;
            // 唤醒所有等待的线程
            notEmpty.signalAll();
            notFull.signalAll();
            System.out.println("消息队列已关闭");
        } finally {
            lock.unlock();
        }
    }

    public void printQueueStatistics() {
        System.out.println("\n=== 消息队列统计 ===");
        System.out.println("当前队列大小: " + queue.size() + "/" + capacity);
        System.out.println("队列状态: " + (shutdown ? "已关闭" : "运行中"));

        System.out.println("\nnotEmpty 条件统计:");
        notEmpty.printDetailedStatistics();

        System.out.println("\nnotFull 条件统计:");
        notFull.printDetailedStatistics();

        // 计算队列效率指标
        double utilization = (double) queue.size() / capacity * 100;
        System.out.printf("队列利用率: %.1f%%%n", utilization);
    }

    public static void main(String[] args) throws InterruptedException {
        MonitoredMessageQueue<String> queue = new MonitoredMessageQueue<>(5);

        // 创建生产者
        Thread producer = new Thread(() -> {
            try {
                for (int i = 1; i <= 10; i++) {
                    String message = "Message-" + i;
                    boolean success = queue.offer(message, 2, TimeUnit.SECONDS);
                    if (!success) {
                        System.out.println("生产者: 添加消息失败 - " + message);
                    }
                    Thread.sleep(100); // 模拟生产间隔
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        // 创建消费者
        Thread consumer = new Thread(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    String message = queue.poll(3, TimeUnit.SECONDS);
                    if (message != null) {
                        System.out.println("消费者: 处理消息 - " + message);
                    } else {
                        System.out.println("消费者: 未获取到消息");
                    }
                    Thread.sleep(150); // 模拟消费时间
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        producer.start();
        consumer.start();

        producer.join();
        consumer.join();

        queue.printQueueStatistics();
        queue.shutdown();
    }
}
